package cc.user;

import java.io.IOException;
import java.io.Reader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;

import compare.SqlJYTUtils;
import oracle.to.mysql.utils.SqlUtils;
/**
 * to data by Thread pool
 *
 */
public class UserInsertEPool {
	
	private static final String mysqlInsertSql= "INSERT INTO `"+SqlUtils.MYSQL_DB+"`.`sys_user_account` (`USER_ID`, `ACCOUNT`, `PASSWORD`, `USER_NAME`,"
			+ " `MAIL`, `PHONE`, `QUESTION`, `ANSWER`, `REGISTER_TIME`, `LOGIN_TIME`, `LOGIN_NUM`, `IS_LOCKED`,"
			+ " `LOCK_USER_ID`, `LOCK_USER_NAME`, `LOCK_TIME`, `LOCK_REASON`, `ISVALID_MAIL`, `VALID_MAIL_TIME`,"
			+ "  `ISVALID_PHONE`, `VALID_PHONE_TIME`, `AAE014`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?, ?,?,?,?,?,?,?,?,?) ";
	
	private static int mysqlcolsSql;
	private static Logger log = Logger.getLogger(" ");
	private static int poolsize = 20;//线程池数量
	private static int pgsize = 500;//单个线程执行的sql数量,oracle分页条数
	private static String MYSQL_TABLE = "sys_user_account";
	static {
		try {
			Class.forName("oracle.jdbc.driver.OracleDriver");
			Class.forName("com.mysql.jdbc.Driver");
		} catch (ClassNotFoundException e) {
			log.error("驱动加载异常：" + e.getMessage());
		}
	}
	
	public static void main(String[] args) throws SQLException, IOException {
		long time = System.currentTimeMillis();
		Connection connOracle = DriverManager.getConnection(SqlUtils.ORACLE_URL, SqlUtils.ORACLE_USER, SqlUtils.ORACLE_PWD);
		getInsertSql(connOracle);
		int count = getCount(connOracle);
		connOracle.close();
		ExecutorService pool = Executors.newFixedThreadPool(poolsize);
		int zheng = count / pgsize;
		int yu = count % pgsize;
		UserInsertEPool r = new UserInsertEPool();
		for (int i = 1; i <= zheng; i++) {
			pool.execute(r.new TransRunnable((i - 1) * pgsize + 1, i * pgsize));
		}
		pool.execute(r.new TransRunnable(count - yu, count));
		pool.shutdown();
		try {
			while (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
				// awaitTermination返回false即超时会继续循环，返回true即线程池中的线程执行完成主线程跳出循环往下执行，每隔10秒循环一次
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		log.info("oracle表,条数："+ count);
		log.info("线程池总数："+ poolsize+","+"子线程数量："+ (zheng+1));
		log.info("子线程执行条数："+ pgsize);
		log.info("全部执行完成，mysql:"+MYSQL_TABLE+"，耗时:"+(System.currentTimeMillis()-time)+"ms");
	}

	private static int getCount(Connection conn) throws SQLException {
		long time = System.currentTimeMillis();
		String sql = "select count(1) from sys_user_E where acb200 is not null and acb200 in(select acb200 from cb20_cc where aab001 in(select aab001 from ab01_cc))";
		Statement stmt = conn.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
		ResultSet rs = stmt.executeQuery(sql);
		int count = 0;
		if (rs.next()) {
			count = rs.getInt(1);
		}
		rs.close();
		stmt.close();
		log.info("查询总数:" + count + "条,耗时：" + (System.currentTimeMillis() - time));
		return count;
	}
	
	private static int checkAccount(Connection connMysql,String account) throws SQLException {
		String sql = "select count(1) from sys_user_account where account='"+account+"'";
		Statement stmt = connMysql.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_UPDATABLE);
		ResultSet rs = stmt.executeQuery(sql);
		int count = 0;
		if (rs.next()) {
			count = rs.getInt(1);
		}
		rs.close();
		stmt.close();
		return count;
	}

	private static void getInsertSql(Connection connOracle) throws SQLException {
		Statement stmt = connOracle.createStatement();
		String querySql = SqlUtils.getSqlByUserE(0, 1);
		ResultSet rs = stmt.executeQuery(querySql);
		ResultSetMetaData mt = rs.getMetaData();
		int cols = mt.getColumnCount();
		rs.close();
		stmt.close();
		mysqlcolsSql = cols-2;//最后一项rn,倒数第二项是acb200
	}
	
	class TransRunnable implements Runnable {
		int start;
		int end;

		public TransRunnable(int start, int end) {
			this.start = start;
			this.end = end;
		}

		@Override
		public void run() {
			Connection connOracle = null;
			Connection connMysql = null;
			Connection connMysqlJYT = null;
			long time = System.currentTimeMillis();
			Statement stmt = null;
			ResultSet rs = null;
			int error = 0;
			int replaceCount = 0;
			try {
				connOracle = DriverManager.getConnection(SqlUtils.ORACLE_URL, SqlUtils.ORACLE_USER, SqlUtils.ORACLE_PWD);
				connMysql = DriverManager.getConnection(SqlUtils.getMySqlUrl());
				connMysqlJYT = DriverManager.getConnection(SqlJYTUtils.getMySqlJytUrl());
				stmt = connOracle.createStatement();
				rs = stmt.executeQuery(SqlUtils.getSqlByUserE(start, end));
				while (rs.next()) {
					String account = rs.getString(2);
					int accountCount = checkAccount(connMysqlJYT, account);
					PreparedStatement stmt2;
					if(accountCount>=1){
						replaceCount++;
						log.error("账户已存在，为"+accountCount+"个，userid:"+rs.getString(1)+",account:"+account);
						stmt2 = connMysql.prepareStatement(mysqlInsertSql.replace("sys_user_account", "sys_user_account_temp")
								.replace("`AAE014`", "`AAE014`,`ACB200`").replace(",?)", ",?,?)"));
						for (int c = 1; c < mysqlcolsSql; c++) {
							insertUser(rs, c, stmt2);
						}
						stmt2.setObject(mysqlcolsSql, "财务系统导入企业");
						stmt2.setObject(mysqlcolsSql+1, rs.getString("ACB200"));
						try {
							stmt2.execute();
						} catch (Exception e) {
							log.error("插入mysql_temp异常,oracle主键："+e.getMessage().replace("Duplicate entry '", "").replace("' for key 'PRIMARY'", "")+",sql:"+stmt2.toString());
							error++;
						}
						stmt2.close();
					}else{
						stmt2 = connMysql.prepareStatement(mysqlInsertSql);
						for (int c = 1; c < mysqlcolsSql; c++) {
							insertUser(rs, c, stmt2);
						}
						stmt2.setObject(mysqlcolsSql, "财务系统导入企业");
						String userid = rs.getString("USERID");
						String acb200 = rs.getString("ACB200");
						Object ctime = rs.getTimestamp("REGISTER_TIME");
						String insertExt = "INSERT INTO `"+SqlUtils.MYSQL_DB+"`.`sys_user_account_ext`"
								+ " (`USER_ID`, `C_TIME`, `TYPE`, `BUSINESSID`, `BUSINESSNAME`, `ID`)"
								+ " VALUES (?, ?, 'E0', ?, '就业职介系统财务导入', ?)";
						PreparedStatement stmt3 = connMysql.prepareStatement(insertExt);
						stmt3.setString(1, userid);
						stmt3.setObject(2, ctime);
						stmt3.setString(3, acb200);
						stmt3.setString(4, userid);
						String accountroleSql="INSERT INTO `"+SqlUtils.MYSQL_DB+"`.`sys_account_role` (`USER_ID`, `ROLE_ID`) VALUES (?, '2FDBA17865607E06E050007F01001D44'), (?, '2FDBA17865607E06E050007F01001D66'), (?, '2FDBA17865617E06E050007F01001D55'),(?, '2FDBA17865617E06E050007F01001D77')";
						PreparedStatement stmt4 = connMysql.prepareStatement(accountroleSql);
						stmt4.setString(1, userid);
						stmt4.setString(2, userid);
						stmt4.setString(3, userid);
						stmt4.setString(4, userid);
						try {
							stmt2.execute();
							stmt3.execute();
							stmt4.execute();
						} catch (Exception e) {
							log.error("插入mysql异常,oracle主键："+e.getMessage().replace("Duplicate entry '", "").replace("' for key 'PRIMARY'", "")+",sql:"+stmt2.toString());
							error++;
						}
						stmt2.close();
						stmt3.close();
						stmt4.close();
					}
				}
			} catch (Exception e) {
				log.error("子线程异常："+e.getMessage());
			} finally {
				try {
					if (stmt != null) {
						stmt.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (rs != null) {
						rs.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connMysql != null) {
						connMysql.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connOracle != null) {
						connOracle.close();
					}
				} catch (SQLException e) {
				}
				try {
					if (connMysqlJYT != null) {
						connMysqlJYT.close();
					}
				} catch (SQLException e) {
				}
				log.info("子线程完成，start:" + start + ",end：" + end + ",用时：" + (System.currentTimeMillis() - time) +"ms,error:"+error+",replaceCount="+replaceCount);
			}
		}
		
		private void insertUser(ResultSet rs,int c,PreparedStatement stmt2) throws SQLException, IOException{
			Object o = rs.getObject(c);
			if (o != null) {
				if (o instanceof oracle.sql.TIMESTAMP) {
					oracle.sql.TIMESTAMP ts = (oracle.sql.TIMESTAMP) o;
					o = new java.sql.Timestamp(ts.timestampValue().getTime());
				} else if (o instanceof java.util.Date) {
					o = new java.sql.Timestamp(((java.util.Date) o).getTime());
				} else if (o instanceof java.sql.Clob) {
					oracle.sql.CLOB clob = (oracle.sql.CLOB) o;
					Reader red = clob.getCharacterStream();
					char[] crs = new char[10240];
					String tmp = "";
					int len;
					while ((len = red.read(crs)) != -1) {
						tmp += new String(crs, 0, len);
					}
					red.close();
					o = new String(tmp);
				}
			}
			stmt2.setObject(c, o);
		}
	}

}